#include "rabbitmq.h"

#include "log.h"

namespace chat {
MQClient::MQClient(const std::string& user, const std::string& password,
                   const std::string& host) {
    loop_ = EV_DEFAULT;

    handler_ = std::make_shared<AMQP::LibEvHandler>(loop_);

    // amqp://root:123456@127.0.0.1:5672/
    std::string url = "amqp://" + user + ":" + password + "@" + host + "/";
    AMQP::Address address(url);
    connection_ =
        std::make_shared<AMQP::TcpConnection>(handler_.get(), address);

    channel_ = std::make_shared<AMQP::TcpChannel>(connection_.get());

    loop_thread_ = std::thread([this]() { ev_run(loop_, 0); });
}

MQClient::~MQClient() {
    ev_async_init(&async_watcher_, watcher_callback);
    ev_async_start(loop_, &async_watcher_);
    ev_async_send(loop_, &async_watcher_);

    loop_thread_.join();
}

void MQClient::watcher_callback(struct ev_loop* loop, ev_async* watcher,
                                int32_t revents) {
    ev_break(loop, EVBREAK_ALL);
}

void MQClient::declareCompoents(const std::string& exchange_name,
                                const std::string& queue_name,
                                const AMQP::ExchangeType exchange_type,
                                const std::string& routing_key) {
    channel_->declareExchange(exchange_name, exchange_type)
        .onError([&exchange_name](const char* message) {
            LOG_ERROR("声明交换机失败：{}: {}", exchange_name, message);
            exit(0);
        })
        .onSuccess([&exchange_name]() {
            LOG_DEBUG("交换机创建成功: {}", exchange_name);
        });

    channel_->declareQueue(queue_name)
        .onError([&queue_name](const char* message) {
            LOG_ERROR("声明队列失败：{}: {}", queue_name, message);
            exit(0);
        })
        .onSuccess([&queue_name]() {
            LOG_DEBUG("test-queue 队列创建成功: {}", queue_name);
        });

    channel_->bindQueue(exchange_name, queue_name, routing_key)
        .onError([&exchange_name, &queue_name](const char* message) {
            LOG_ERROR("{}<->{} 绑定失败：{}", exchange_name, queue_name,
                      message);
            exit(0);
        })
        .onSuccess([&exchange_name, &queue_name]() {
            LOG_DEBUG("{}<->{} 绑定成功！", exchange_name, queue_name);
        });
}

bool MQClient::publish(const std::string& exchange_name,
                       const std::string& routing_key,
                       const std::string& message) {
    bool ret = channel_->publish(exchange_name, routing_key, message);
    if (ret == false) {
        LOG_ERROR("publish失败: {}<->{}: {}", exchange_name, routing_key,
                  message);
    }
    return ret;
}

bool MQClient::consume(const std::string& queue_name,
                       const std::string& consume_tag,
                       const MessageCallback_t callback) {
    channel_->consume(queue_name, consume_tag)
        .onReceived([this, callback](const AMQP::Message& message,
                                     uint64_t deliveryTag, bool redelivered) {
            callback(message.body(), message.bodySize());
            channel_->ack(deliveryTag);  // 对消息进行确认
        })
        .onError([&queue_name](const char* message) {
            LOG_ERROR("订阅队列消息失败: {}: {}", queue_name, message);
            exit(0);
        });

    return true;
}
}  // namespace chat
